package com.zdgk.core.mqtt;

import com.zdgk.core.control.config.MqttConfigInfo;
import com.zdgk.core.web.service.ConfigService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.UUID;

@Slf4j
@Configuration
public class MqttClientConfig {

    @Resource
    private ConfigService configService;

    private static MqttClient client;
    private MqttConfigInfo mqttConfigInfo;
    private int timeout = 2;
    private int keepalive = 60;

    @PostConstruct
    public void init() {
        try {
            mqttConfigInfo = configService.getMqttKey();
            if (isConnected()) {
                client.disconnect();
            }
            client = null;
            this.connect();
        } catch (Exception e) {
            log.error("mqtt连接失败");
        }
    }

    public static MqttClient getClient() {
        return client;
    }

    public static void setClient(MqttClient client) {
        MqttClientConfig.client = client;
    }

    /**
     * 设置mqtt连接参数
     *
     * @param username
     * @param password
     * @param timeout
     * @param keepalive
     * @return
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        return options;
    }

    public boolean isConnected() {
        return client != null && client.isConnected();
    }

    /**
     * 连接mqtt服务端，得到MqttClient连接对象
     */
    public void connect() throws MqttException {
        if (client == null) {
            client = new MqttClient(mqttConfigInfo.getMqttHost(), mqttConfigInfo.getMqttClientId()+ UUID.randomUUID(), new MemoryPersistence());
            client.setCallback(new MqttCallback(MqttClientConfig.this,configService));
        }
        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(mqttConfigInfo.getMqttUsername() == null ? "" : mqttConfigInfo.getMqttUsername(), mqttConfigInfo.getMqttPassword() == null ? "" : mqttConfigInfo.getMqttPassword(), timeout, keepalive);
        if (!client.isConnected()) {
            client.connect(mqttConnectOptions);
        } else {
            client.disconnect();
            client.connect(mqttConnectOptions);
        }
    }

    public void publishUpload(String pushMessage) {
        publish(pushMessage, mqttConfigInfo.getMqttUploadTopic(), 0, false);
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic) {
        publish(pushMessage, topic, 0, false);
    }

    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, int qos, boolean retained) {
        if (isConnected()) {
            MqttMessage message = new MqttMessage();
            message.setPayload(pushMessage.getBytes());
            message.setQos(qos);
            message.setRetained(retained);
            MqttTopic mqttTopic = MqttClientConfig.getClient().getTopic(topic);
            if (null == mqttTopic) {
            }
            MqttDeliveryToken token;//Delivery:配送
            synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁
                try {
                    token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                    token.waitForCompletion(1000L);
                } catch (MqttPersistenceException e) {
                    e.printStackTrace();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            if (isConnected()) {
                MqttClientConfig.getClient().subscribe(topic, qos);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */
    public void cleanTopic(String topic) {
        if (isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            System.out.println("取消订阅失败！");
        }
    }
}
